package cn.wen233.iocdemo.infrustructure.util;

import cn.hutool.core.util.ObjectUtil;
import cn.wen233.iocdemo.domain.enums.LedState;
import cn.wen233.iocdemo.domain.enums.MotorState;
import cn.wen233.iocdemo.infrustructure.config.AppConfig;
import cn.wen233.iocdemo.infrustructure.config.MQTTProperties;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;

import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.CompletableFuture;

/**
 * mqtt协议处理工具类
 *
 * @author wenei
 * @date 2021-06-23 10:08
 */
public class MqttChannel implements AutoCloseable {

    private static final Logger log = LoggerFactory.getLogger(MqttChannel.class);

    /**
     * 客户机id
     */
    private static final String clientId = "http-consumer";

    /**
     * 服务质量
     */
    private static final int qos = 1;

    private static String ip = IocUtil.getBean(MQTTProperties.class).getIp();

    private static int port = IocUtil.getBean(MQTTProperties.class).getPort();

    private static final String serviceUrl = String.format("tcp://%s:%d", ip, port);

    private static final Map<String, String> topicMap = new HashMap<>();

    /**
     * 默认订阅的主题
     */
    private static final List<String> defaultSubscribeTopic = Collections.singletonList("dht11Topic");

    private static final Map<String, String> defaultTopicStatus = new HashMap<>();

    static {
        final String topic = "51Topic";
        defaultTopicStatus.put(topic, LedState.OFF.getCommand());
        defaultTopicStatus.put(topic, MotorState.OFF.getCommand());
    }

    private static MqttClient mqttClient = null;

    static {
        // 内存存储
        MemoryPersistence persistence = new MemoryPersistence();
        // 创建客户端
        try {
            mqttClient = new MqttClient(serviceUrl, clientId, persistence);
            // 创建链接参数
            MqttConnectOptions connOpts = new MqttConnectOptions();
            // 在重新启动和重新连接时记住状态
            connOpts.setCleanSession(false);
            // 设置连接的用户名
//            connOpts.setUserName(userName);
//            connOpts.setPassword(password.toCharArray());
            // 建立连接
            mqttClient.connect(connOpts);
            // 设置回调函数
            mqttClient.setCallback(new MqttCallback() {

                @Override
                public void connectionLost(Throwable cause) {
                    log.info("失去连接");
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    log.info("Time: {}, Topic:{}, Message:{}",
                            new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()), topic, message);
                    topicMap.put(topic, new String(message.getPayload()));
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    log.info("更新订阅:" + token.isComplete());
                }

            });
            // 默认订阅的主题
            defaultSubscribeTopic.forEach(x -> {
                try {
                    mqttClient.subscribe(x, qos);
                } catch (MqttException e) {
                    e.printStackTrace();
                }
                topicMap.put(x, "temperature:-99.00000, humidity:-99.00000");
            });
            // 默认51控制主题状态设定
            defaultTopicStatus.forEach(MqttChannel::publish);
        } catch (MqttException mqttException) {
            mqttException.printStackTrace();
        }
    }

    /**
     * 异步调用推送方法
     */
    public static void asyncPublish(String topic, String sendData) {
        CompletableFuture.supplyAsync(() -> {
            publish(topic, sendData);
            return null;
        });
    }

    public static void publish(String topic, String sendData) {
        try {
            // 创建消息
            MqttMessage message = new MqttMessage(sendData.getBytes(StandardCharsets.UTF_8));
            // 设置消息的服务质量
            message.setQos(qos);
            // 发布消息
            mqttClient.publish(topic, message);
            log.info("向 {} 的 {} 主题发送:{}", serviceUrl, topic, sendData);
        } catch (MqttPersistenceException e) {
            e.printStackTrace();
        } catch (MqttSecurityException e) {
            e.printStackTrace();
        } catch (MqttException e) {
            e.printStackTrace();
        }

    }

    public static String subscribe(String topic) {
        return topicMap.get(topic);
    }

    @Override
    public void close() throws Exception {
        mqttClient.disconnect();
        mqttClient.close();
    }
}
